package org.example.flink.operation;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

public class Hello18ProcessFunction {
    public static void main(String[] args) throws Exception {
        //执行环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        //获取数据源
        DataStreamSource<String> source = environment.readTextFile("data/wordcount1.txt");
        //数据处理
        source.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String line, ProcessFunction<String, String>.Context context, Collector<String> collector)
                    throws Exception {
                String[] words = line.split(" ");
                for (String word : words) {
                    collector.collect(word);
                }
            }
        }).process(new ProcessFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void processElement(String word, ProcessFunction<String, Tuple2<String, Integer>>.Context context,
                                       Collector<Tuple2<String, Integer>> collector) throws Exception {
                collector.collect(Tuple2.of(word, 1));
            }
        }).keyBy(tuple2 -> tuple2.f0).process(new KeyedProcessFunction<String, Tuple2<String, Integer>, Object>() {
            @Override
            public void processElement(Tuple2<String, Integer> tuple2, KeyedProcessFunction<String,
                    Tuple2<String, Integer>, Object>.Context context, Collector<Object> collector) throws Exception {
                System.out.println(System.currentTimeMillis() +  " processElement[" + context.getCurrentKey() + "]");
                collector.collect(tuple2);
            }
        }).print();

        //运行环境
        environment.execute();
    }
}
